package com.ztrue.framework.disruptor;

import com.lmax.disruptor.*;
import com.ztrue.framework.infrastructure.IClone;
import com.ztrue.framework.infrastructure.IDistribute;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 消息工厂
 *
 * @author zhangw
 * @date 2016/11/30 23:29
 * @version: 1.0
 */
public class MessageQueueFactory {
    public static <M extends IDistribute & IClone> IMessageQueue<M> make(MessageHandler<M> eventHandler, int bufferSize, int threadNumbers, EventFactory<M> eventFactory) {
        //环形队列
        RingBuffer<M> ringBuffer = RingBuffer.createMultiProducer(eventFactory, bufferSize, new YieldingWaitStrategy());
        //创建线程池
        ExecutorService executors = Executors.newFixedThreadPool(threadNumbers);
        //创建SequenceBarrier
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        //创建消息处理器
        BatchEventProcessor<M> eventProcessor = new BatchEventProcessor<M>(ringBuffer, sequenceBarrier, eventHandler);
        ringBuffer.addGatingSequences(eventProcessor.getSequence());
        //把消息处理器提交到线程池
        executors.submit(eventProcessor);
        return new MessageQueue<M>(ringBuffer, executors);
    }
}
